package kevin.hbase.util;

import cc.chinagps.gateway.buff.DeliverBuff;
import kevin.hbase.common.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;

@Slf4j
public class HbaseUtils {
    private static Connection connection;

    static {
        try {
            Configuration config = new Configuration();
            config.addResource(new Path("conf/hbase-site.xml"));
            config = HBaseConfiguration.create(config);
            connection = ConnectionFactory.createConnection(config);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    public static Connection getConnection() {
        return connection;
    }

    private static byte[] family_loginout = "loginout".getBytes();
    private static byte[] qualifier_base_loginout = "data".getBytes();

    public static int getGwId(String callLetter) throws Exception {
        Table table = connection.getTable(TableName.valueOf("t_unitloginout"));
        byte[] startKey = HBaseKeyUtil.getKey(callLetter, System.currentTimeMillis() + 24 * 3600000L);
        byte[] endKey = HBaseKeyUtil.getKey(callLetter, 24L * 3600000L * 42 * 365);
        Scan scan = new Scan(startKey, endKey);
        ResultScanner rsscan = table.getScanner(scan);
        scan.setCaching(2);
        Iterator<Result> iter = rsscan.iterator();
        if (iter.hasNext()) {
            Result rs = iter.next();
            rs.getRow();
            byte[] baseData = rs.getValue(family_loginout, qualifier_base_loginout);
            DeliverBuff.DeliverUnitLoginOut deliverUnitLoginOut = DeliverBuff.DeliverUnitLoginOut.parseFrom(baseData);
            int gwId = deliverUnitLoginOut.getGatewayId();
            //int flag = deliverUnitLoginOut.getInorout();
            if (gwId != 0) {
                return gwId;
            }
        }
        return 0;
    }

    /**
     * @param callLetter
     * @param startTime
     * @param endTime
     * @param order      Gps数据根据gpsTime的排序方式，由于key的设计，最新时间的数据在最前面，即end对应的数据在前面，start对应的数据在后面
     *                   注：hbase中rowkey的存储顺序是按照rowkey的ASCAII值升序排序
     * @return
     * @throws Exception
     */
    public static ResultScanner getScanner(String callLetter, long startTime, long endTime, Order order) throws Exception {
        Connection connection = HbaseUtils.getConnection();
        Table table = connection.getTable(TableName.valueOf("t_history"));
        byte[] startRow = HBaseKeyUtil.getKey(callLetter, endTime);
        byte[] endRow = HBaseKeyUtil.getKey(callLetter, startTime);
        Scan scan;
        if (order == Order.ASC) {
            scan = new Scan(endRow, startRow);
            scan.setReversed(true);//启用倒序扫描，即从rowKey较大的开始读取数据
        } else {
            scan = new Scan(startRow, endRow);
        }
        scan.setCaching(2);
        ResultScanner rsscan = table.getScanner(scan);
        table.close();
        return rsscan;
    }

    private static byte[][] getSplitKeys() {
        int start = 48;
        int end = 111;
        byte[][] splitKeys = new byte[end - start][];
        TreeSet<byte[]> rows = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);// 升序排序
        for (int i = 0; i < end - start; i++) {
            rows.add(new byte[]{(byte) (start + 1 + i)});
        }
        Iterator<byte[]> rowKeyIter = rows.iterator();
        int i = 0;
        while (rowKeyIter.hasNext()) {
            byte[] tempRow = rowKeyIter.next();
            rowKeyIter.remove();
            splitKeys[i] = tempRow;
            i++;
        }
        return splitKeys;
    }

    public static boolean createTableBySplitKeys(String tableName, List<String> columnFamilys, Connection connection) {
        try {
            if (StringUtils.isBlank(tableName) || columnFamilys == null || columnFamilys.size() < 0) {
                log.error("Parameters tableName|columnFamily should not be null,Please check!");
                return false;
            }
            Admin admin = connection.getAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                return true;
            } else {
                HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
                for (String cf : columnFamilys) {
                    HColumnDescriptor column = new HColumnDescriptor(cf);
                    column.setCompressionType(Compression.Algorithm.SNAPPY);
                    tableDescriptor.addFamily(column);
                }
                byte[][] splitKeys = getSplitKeys();
                admin.createTable(tableDescriptor, splitKeys);// 指定splitkeys
                log.info("Create Table " + tableName + " Success!columnFamily:" + columnFamilys.toString() + "===");
            }
        } catch (MasterNotRunningException e) {
            log.error(e.getMessage(), e);
            return false;
        } catch (ZooKeeperConnectionException e) {
            log.error(e.getMessage(), e);
            return false;
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            return false;
        }
        return true;
    }
}
